package test.mq;

import java.io.UnsupportedEncodingException;
import java.util.List;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;


import hyl.base.mq.rocket.MyRocketMq;
import hyl.core.MyDate;
import hyl.core.info.CMsg;
import hyl.core.run.IFinishT;

public class testrocket_receive {
	
	public testrocket_receive() {
		mqs = MyRocketMq.iniConsumer("consumer_group1","group2",1);
	}
	
	MyRocketMq mqs = null;

	public static void main(String[] args) throws Exception {
		testrocket_receive t = new testrocket_receive();
		t.test();
		//testconn();
	}
	
	public void test() {
		System.out.println("启动接收器");
		mqs.set主题("WS_CMD").set回调函数( if2);
		mqs.receive();
	}

	IFinishT<CMsg> if2=new IFinishT<CMsg>() {

		@Override
		public boolean success(CMsg o) {
			System.out.println("接收"+o.toJsonString());
			MyDate.sleep(2000);
			System.out.println("处理完毕"+o.s键);
			return false;
		}

		@Override
		public void error(CMsg o, Throwable e) {
			
		}
		
	};

	public static void testconn() throws MQClientException {
		DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("rmq-group");
		consumer.setNamesrvAddr("192.168.0.147:9876;192.168.0.147:9877");
		consumer.setInstanceName("consumer");
		consumer.subscribe("itmayiedu-topic1", "TagA");
		consumer.subscribe("itmayiedu-topic2", "TagA");
		consumer.registerMessageListener(new MessageListenerConcurrently() {

			@Override
			public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
				for (MessageExt msg : msgs) {
					System.out.println(msg.getMsgId() + "---" + new String(msg.getBody()));
				}
				return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
			}
		});
		consumer.start();
		System.out.println("开始消费...");

	}

	private MessageListenerConcurrently listener = new MessageListenerConcurrently() {
		@Override
		public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
			try {
				for (MessageExt msg : msgs) {
					String topic = msg.getTopic();
					String msgBody = new String(msg.getBody(), "utf-8");
					String tags = msg.getTags();
					System.out.println("收到消息： topic:" + topic + " ,tags:" + tags + " ,msg: " + msgBody);
				}
			} catch (UnsupportedEncodingException e) {
				e.printStackTrace();
			}
			return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
		}
	};
}
